package com.zlx.rulemk.marketing.functions;

import com.alibaba.fastjson.JSON;
import com.zlx.rulemk.marketing.beans.*;
import com.zlx.rulemk.marketing.metrics.RuleIsMatchAvgTimeGauge;
import com.zlx.rulemk.marketing.service.RulelMatchServiceImpl;
import com.zlx.rulemk.marketing.utils.FlinkStateDescUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.shaded.guava18.com.google.common.base.Stopwatch;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-14
 * @Description: TODO
 * @Version: 1.0
 */
@Slf4j
public class RuleMatchKeyedProcessFunctionV2 extends KeyedProcessFunction<String, DynamicKeyedBean, RuleMatchResult> {

    // flink缓存的明细数据
    transient ListState<EventBean> listState;

    // Flink执行规则对象 flink
    // key：rule
    transient ListState<RuleStateBean> ruleState;

    // 触发规则的计算类
    RulelMatchServiceImpl rulelMatchServiceImpl;

    //String 是分组的key，不是具体的value，将来获取对应的规则执行
    ListState<Tuple2<String, Long>> timerInfoState;

    Counter rule_calc_timeAmount;
    Counter rule_match_calc_timeAmount;
    Counter rule_not_match_calc_timeAmount;
    Counter rule_calc_cntAmount;
    Counter rule_is_match_calc_cntAmount;
    Counter rule_not_match_calc_cntAmount;

    RuleIsMatchAvgTimeGauge ruleIsMatchAvgTimeGauge;

    @Override
    public void open(Configuration parameters) throws Exception {

        listState = getRuntimeContext().getListState(FlinkStateDescUtils.getEventBeansDesc());

        ruleState = getRuntimeContext().getListState(FlinkStateDescUtils.ruleStateDesc2);

        rulelMatchServiceImpl = new RulelMatchServiceImpl(listState);

        // 记录规则定时信息的state
        timerInfoState = getRuntimeContext().getListState(FlinkStateDescUtils.ruleTimerStateDesc);

        rule_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_timeAmount");
        rule_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_timeAmount");
        rule_not_match_calc_timeAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_timeAmount");
        rule_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_calc_cntAmount");
        rule_is_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_is_match_calc_cntAmount");
        rule_not_match_calc_cntAmount = getRuntimeContext().getMetricGroup().counter("rule_not_match_calc_cntAmount");

        ruleIsMatchAvgTimeGauge = getRuntimeContext().getMetricGroup().gauge("计算成立的规则的平均每次计算耗时",  new RuleIsMatchAvgTimeGauge());

    }


    /**
     * Flink框架 日志数据驱动
     * @param dynamicKeyedBean
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(DynamicKeyedBean dynamicKeyedBean, Context ctx, Collector<RuleMatchResult> out)
            throws Exception {

        // 将日志数据添加到flinkState中
        listState.add(dynamicKeyedBean.getEventBean());

        // 设置规则
        setRuleState(dynamicKeyedBean.getCdcLogBean());

        RuleStateBean ruleStateBean = null;
        Iterator<RuleStateBean> iterator = ruleState.get().iterator();
        while (iterator.hasNext()){
            ruleStateBean = iterator.next();
        }
        if (ruleStateBean == null){
            log.error("系统没有对应的规则，程序异常，ruleName:{},具体的日志：{}",dynamicKeyedBean.getRuleName(),dynamicKeyedBean.getEventBean());
            return ;
        }

        MarketingRule marketingRule = ruleStateBean.getMarketingRule();
        KieSession kieSession = ruleStateBean.getKieSession();
        /**
         * 第一次匹配的时候  timerMatchDroolsFact 为null 先判断【触发条件，画像条件，行为组合条件】如果满足，在注册定时器条件
         */
        RuleMatchFact ruleMatchFact = new RuleMatchFact(marketingRule, rulelMatchServiceImpl,
                dynamicKeyedBean.getEventBean(), false,null);

        // 创建stopwatch并开始计时
        Stopwatch stopwatch = Stopwatch.createStarted();

        log.debug("调用drools引擎，开始执行计算");

        // 调用drools引擎，执行计算
        kieSession.insert(ruleMatchFact);
        kieSession.fireAllRules();

        // 获取计算结果
        boolean matchResult = ruleMatchFact.isMatchResult();
        log.debug("==============调用drools引擎，结束计算,计算结果：{},耗时{}ms,动态分组的key：{},分组的值：{}，规则rule_name：{}",
                matchResult,(stopwatch.elapsed(TimeUnit.MILLISECONDS)),
                dynamicKeyedBean.getKeyNames(),dynamicKeyedBean.getKeyValue(),
                marketingRule.getRuleId()
        );

        rule_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
        if(matchResult){
            rule_match_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
            rule_is_match_calc_cntAmount.inc();

            ruleIsMatchAvgTimeGauge.incRuleIsMatchCount();
            ruleIsMatchAvgTimeGauge.incRuleIsMatchTimeAmount(stopwatch.elapsed(TimeUnit.MILLISECONDS));

        }else{
            rule_not_match_calc_timeAmount.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS));
            rule_not_match_calc_cntAmount.inc();
        }

        // 判断计算结果是否为true, 并继续判断是否有定时条件
        if(matchResult){
            // 再判断这个规则是否是一个带定时条件的规则
            if (marketingRule.isOnTimer()) {  // 是带定时条件的规则
                // 从规则中取出所有定时条件
                List<TimerCondition> timerConditionList = marketingRule.getTimerConditionList();
                //TODO 目前限定一个规则中只有一个时间条件
                TimerCondition timerCondition = timerConditionList.get(0);
                // 注册定时器
                long triggerTime = dynamicKeyedBean.getEventBean().getTimeStamp() + timerCondition.getTimeLate();
                ctx.timerService().registerEventTimeTimer(triggerTime);

                long currentWatermark = ctx.timerService().currentWatermark();

                // 在 定时信息state中进行记录
                timerInfoState.add(Tuple2.of(dynamicKeyedBean.getKeyNames(), triggerTime));

                log.debug("注册定时器规则，动态的key：{},触发时间：{},当前系统的时间:{}，定时规则条件：{}",ctx.getCurrentKey(),
                        triggerTime,currentWatermark,ruleStateBean.getMarketingRule().getTimerConditionList());

            } else {  // 不带定时条件
                // 输出规则成立的结果
                out.collect(new RuleMatchResult(dynamicKeyedBean.getKeyValue(),marketingRule.getRuleId(),dynamicKeyedBean.getEventBean().getTimeStamp(),System.currentTimeMillis()));
            }
        }

    }

    private void setRuleState(CDCLogBean cdcLogBean) throws Exception {

        if(cdcLogBean == null) return;

        // 规则平台上操作的规则的数据
        List<RuleTableRecord> ruleTableRecords = cdcLogBean.getData();
        if(ruleTableRecords == null ) return;

        // 规则平台上的操作的类型
        String operationType = cdcLogBean.getType();

        // 根据不同的操作类型，去操作state中的规则数据
        for (RuleTableRecord ruleTableRecord : ruleTableRecords) {

            // 创建stopwatch并开始计时
            Stopwatch stopwatch = Stopwatch.createStarted();

            String rule_condition_json = ruleTableRecord.getRule_condition_json();
            String rule_controller_drl = ruleTableRecord.getRule_controller_drl();
            String rule_name = ruleTableRecord.getRule_name();
            MarketingRule marketingRule = JSON.parseObject(rule_condition_json, MarketingRule.class);

            RuleStateBean ruleStateBean = null;
            Iterator<RuleStateBean> iterator = ruleState.get().iterator();
            while (iterator.hasNext()){
                ruleStateBean = iterator.next();
                break;
            }

            if ("INSERT".equals(operationType) || ("UPDATE".equals(operationType) && "1".equals(ruleTableRecord.getRule_status()))) {

                // 往该list中放入一个新的  规则bean对象

                // 往statebean中放kiesession
                KieHelper kieHelper = new KieHelper();
                kieHelper.addContent(rule_controller_drl, ResourceType.DRL);
                KieSession kieSession = kieHelper.build().newKieSession();
                RuleStateBean newRuleStateBean = new RuleStateBean(marketingRule, kieSession);

                if(ruleStateBean == null){
                    ruleState.add(newRuleStateBean);
                }else{
                    // 先移除之前的 在新增一个
                    Iterator<RuleStateBean> iterator1 = ruleState.get().iterator();
                    while (iterator1.hasNext()){
                        iterator1.next();
                        iterator1.remove();
                        break;
                    }
                    ruleState.add(newRuleStateBean);
                }
                log.debug(">>>>初始化规则成功，规则名称：{},耗时：{} ms",rule_name,stopwatch.elapsed(TimeUnit.MILLISECONDS));
            }
            if ("DELETE".equals(operationType) || ("UPDATE".equals(operationType) && "0".equals(ruleTableRecord.getRule_status()))) {

                Iterator<RuleStateBean> iterator2 = ruleState.get().iterator();
                while(iterator2.hasNext()){
                    iterator2.remove();
                    break;
                }
            }

        }


    }
}
